/**
 * 
 */
package xiaojian.toolkit.fsm;

import java.lang.reflect.InvocationTargetException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections4.Closure;
import org.apache.commons.collections4.Predicate;
import org.apache.commons.collections4.Transformer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xiaojian.toolkit.ebus.EventBus;
import xiaojian.toolkit.fsm.event.TimeoutEvent;
import xiaojian.toolkit.fsm.statistic.FSMStatisticer;

/**
 * @author hp
 *
 */
public class DefaultFSM implements FiniteStateMachine {

	private static final Logger logger = LoggerFactory
			.getLogger(DefaultFSM.class);

	private Map<Object, MutableFSMContext> contextHolder = new ConcurrentHashMap<Object, MutableFSMContext>();

	private String name;
	private int fsmId;
	private ContextKeyFactory keyFactory;
	private Predicate initeventAcceptor;
	private FSMContextFactory contextFactory;
	private StateRepository stateRepository;
	private ExecutorService[] execServices = null;
	private int executorPoolSize = 1;
	private AtomicInteger serviceIdx = new AtomicInteger(0);
	private EventBus eventBus;
	private String eventStateChanged;
	private boolean logStateChanged = true;
	private AtomicInteger ctxCount = new AtomicInteger(0);
	private boolean checkBusy = true;
	private int busyThreshold = 1000;
	private Closure busyHandler = null;
	private Transformer snapshotFactory = null;
	private AtomicInteger handledTransactionCount = new AtomicInteger(0);
	private AtomicInteger bypassTransactionCount = new AtomicInteger(0);
	private FSMStatisticer statisticer;
	private ScheduledExecutorService timerExecService = null;
	private Map<TimeoutEvent, Long> timerEvents = new HashMap<TimeoutEvent, Long>();
	private long timerEventCheckInterval = 1000;

	public Transformer getSnapshotFactory() {
		return snapshotFactory;
	}

	public void setSnapshotFactory(Transformer snapshotFactory) {
		this.snapshotFactory = snapshotFactory;
	}

	public EventBus getEventBus() {
		return eventBus;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public String getEventStateChanged() {
		return eventStateChanged;
	}

	public void setEventStateChanged(String eventStateChanged) {
		this.eventStateChanged = eventStateChanged;
	}

	public String getName() {
		return name;
	}

	public DefaultFSM(String name, int id) {
		this.name = name;
		this.fsmId = id;
	}

	private ExecutorService createServiceOf(final int idx) {
		return Executors.newFixedThreadPool(1, new ThreadFactory() {

			public Thread newThread(Runnable r) {
				return new Thread(r, name + "-" + idx);
			}
		});
	}

	public void start() {
		execServices = new ExecutorService[executorPoolSize];

		for (int idx = 0; idx < executorPoolSize; idx++) {
			execServices[idx] = createServiceOf(idx);
		}
		timerExecService = new ScheduledThreadPoolExecutor(1,
				new ThreadFactory() {

					public Thread newThread(Runnable r) {
						return new Thread(r, name + "-timer");
					}
				});
		scheduleCheckTimeoutEvents();
	}

	public void stop() {
		timerExecService.shutdownNow();

		for (ExecutorService service : execServices) {
			service.shutdownNow();
		}

		execServices = null;
	}

	public int getExecutorPoolSize() {
		return executorPoolSize;
	}

	public void setExecutorPoolSize(int poolSize) {
		this.executorPoolSize = poolSize;
	}

	public String getStateRepositoryAsString() {
		return stateRepository.toString();
	}

	public StateRepository getStateRepository() {
		return stateRepository;
	}

	public void setStateRepository(StateRepository stateRepository) {
		this.stateRepository = stateRepository;
	}

	public FSMContextFactory getContextFactory() {
		return contextFactory;
	}

	public String getContextFactoryAsString() {
		return contextFactory.toString();
	}

	public void setContextFactory(FSMContextFactory contextFactory) {
		this.contextFactory = contextFactory;
	}

	public ContextKeyFactory getKeyFactory() {
		return keyFactory;
	}

	public void setKeyFactory(ContextKeyFactory keyFactory) {
		this.keyFactory = keyFactory;
	}

	public int getContextCount() {
		return ctxCount.get();
	}

	public String[] getPendingContextDetail() {
		Set<Map.Entry<Object, MutableFSMContext>> ctxs = contextHolder
				.entrySet();

		List<String> ret = new ArrayList<String>();
		SimpleDateFormat formater = new SimpleDateFormat("yyMMdd-HH:mm:ss.SSS");

		for (Map.Entry<Object, MutableFSMContext> entry : ctxs) {
			StringBuilder sb = new StringBuilder();

			sb.append(entry.getKey().toString());
			sb.append(":[");

			FSMContext ctx = entry.getValue();
			sb.append(ctx.getCurrentState());
			sb.append("]create:");
			sb.append(formater.format(new Date(ctx.getCreateTime())));
			sb.append("/last:");
			sb.append(formater.format(new Date(ctx.getLastModify())));
			ret.add(sb.toString());
		}

		return ret.toArray(new String[0]);
	}

	private int getAndIncServiceIndex() {
		// decide which idx to get session
		return (serviceIdx.getAndIncrement()) % execServices.length;
	}

	private ExecutorService selectService() {
		return execServices[getAndIncServiceIndex()];
	}

	private void removeContext(MutableFSMContext ctx) {
		if (null != this.statisticer) {
			try {
				statisticer.incHandledTransactionEnd(this.name, this.fsmId,
						ctx.getEndReason());
			} catch (Exception e) {
				logger.error("[" + name + "]incHandledTransactionEnd:", e);
			}
		}

		Object key = ctx.getKey();

		this.keyFactory.removeKey(key);
		if (null != this.contextHolder.remove(key)) {
			// 已做移除操作
			this.ctxCount.decrementAndGet();
		}

		try {
			ctx.destroy();
		} catch (Exception e) {
			logger.error("[" + name + "]removeContext:", e);
		}
	}

	private State getStateOf(String stateName) {
		return this.stateRepository.find(stateName);
	}

	private void fireStateChanged(FSMContext ctx, String fromState,
			String toState, Object event) {
		if (logStateChanged && null != eventBus && null != eventStateChanged) {
			Object snapshot = ctx.getSnapshot();

			if (null != snapshot) {
				try {
					PropertyUtils.setProperty(snapshot, "fsm", name);
					PropertyUtils.setProperty(snapshot, "fromState", fromState);
					PropertyUtils.setProperty(snapshot, "toState", toState);
					PropertyUtils.setProperty(snapshot, "eventRecved",
							null != event ? event.toString() : "");
					PropertyUtils.setProperty(snapshot, "time", new Date());
				} catch (IllegalAccessException e) {
					logger.error("fireStateChanged:", e);
				} catch (InvocationTargetException e) {
					logger.error("fireStateChanged:", e);
				} catch (NoSuchMethodException e) {
					logger.error("fireStateChanged:", e);
				}

				eventBus.fireEvent(this.eventStateChanged, snapshot);
			}
		}
	}

	private void initSnapshot(MutableFSMContext ctx) {
		if (this.logStateChanged && (null != this.snapshotFactory)) {
			try {
				ctx.setSnapshot(snapshotFactory.transform(ctx));
			} catch (Exception e) {
				logger.error("[" + name + "]initSnapshot:", e);
			}
		}
	}

	private void initContext(MutableFSMContext ctx, Object event) {
		State state = getStateOf(ctx.getCurrentState());
		if (null == state) {
			logger.warn("Can not found valid state for ["
					+ ctx.getCurrentState() + "] , remove context [" + ctx
					+ "]");
			removeContext(ctx);
		} else {

			initSnapshot(ctx);

			boolean ret = false;

			try {
				ret = state.enter(this, ctx);
			} catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".enter:", e);
			}

			if (!ret) {
				if (logger.isDebugEnabled()) {
					logger.debug("enter return false, ctx [" + ctx
							+ "] will end normally.");
				}
				removeContext(ctx);
				return;
			}
			fireStateChanged(ctx, "", ctx.getCurrentState(), event);

			initSnapshot(ctx);

			dispatchEvent(ctx, event);
		}
	}

	private void dispatchEvent(MutableFSMContext ctx, Object event) {
		State currentState = getStateOf(ctx.getCurrentState());
		if (null == currentState) {
			logger.error("Internal Error: Can not found valid state for ["
					+ ctx.getCurrentState() + "] , remove context [" + ctx
					+ "]");
			removeContext(ctx);
		}

		String nextStateName = null;

		try {
			nextStateName = currentState.acceptEvent(this, ctx, event);
		} catch (Exception e) {
			logger.error(ctx.getCurrentState() + ".acceptEvent:", e);
		}

		if (null == nextStateName) {
			// this context's FSM state reach end
			try {
				currentState.leave(this, ctx);
			} catch (Exception e) {
				logger.error(ctx.getCurrentState() + ".leave:", e);
			}

			if (logger.isDebugEnabled()) {
				logger.debug("ctx [" + ctx + "] will end normally.");
			}

			fireStateChanged(ctx, ctx.getCurrentState(), "", event);
			removeContext(ctx);
		} else if (nextStateName.equals(ctx.getCurrentState())) {
			// no change
		} else {
			State nextState = getStateOf(nextStateName);
			if (null == nextState) {
				logger.error("Internal Error: Can not found valid state for ["
						+ nextStateName + "] , just ignore");
			} else {
				// next state is valid
				try {
					currentState.leave(this, ctx);
				} catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".leave:", e);
				}

				String prevStateName = ctx.getCurrentState();
				ctx.setCurrentState(nextStateName);

				boolean ret = false;
				try {
					ret = nextState.enter(this, ctx);
				} catch (Exception e) {
					logger.error(ctx.getCurrentState() + ".enter:", e);
				}

				fireStateChanged(ctx, prevStateName, ctx.getCurrentState(),
						event);
				initSnapshot(ctx);

				if (!ret) {
					if (logger.isDebugEnabled()) {
						logger.debug("enter return false, ctx [" + ctx
								+ "] will end normally.");
					}

					removeContext(ctx);
				}
			}
		}
	}

	private void incBypassTransaction() {
		bypassTransactionCount.incrementAndGet();
		if (null != this.statisticer) {
			try {
				statisticer.incBypassTransaction(this.name, this.fsmId);
			} catch (Exception e) {
				logger.error("[" + name + "]incBypassTransaction:", e);
			}
		}
	}

	private void incHandledTransactionStart() {
		handledTransactionCount.incrementAndGet();
		if (null != this.statisticer) {
			try {
				statisticer.incHandledTransactionStart(this.name, this.fsmId);
			} catch (Exception e) {
				logger.error("[" + name + "]incHandledTransactionStart:", e);
			}
		}
	}

	private boolean handleMorethanLimit(Object event) {
		if (this.checkBusy && this.getContextCount() >= this.busyThreshold) {

			incBypassTransaction();

			// 超过CTX限制值
			if (null != busyHandler) {
				try {
					busyHandler.execute(event);
				} catch (Exception e) {
					logger.error(busyHandler.toString() + ".execute:", e);
				}
			} else {
				logger.error("reach ctx limit, and missing BusyHandler.");
			}
			return true;
		}
		return false;
	}

	public void acceptEvent(final Object event) {
		Object key = keyFactory.transformToKey(event);

		if (null == key) {
			logger.warn("event " + event
					+ " can not transform key, just ignore");
			return;
		}
		// lookup key in contextHolder
		final MutableFSMContext ctx = contextHolder.get(key);
		if (null == ctx) {
			if (!this.initeventAcceptor.evaluate(event)) {
				if (logger.isDebugEnabled()) {
					logger.debug("event [" + event
							+ "] not acceptable, just ignore");
				}
				return;
			}

			if (handleMorethanLimit(event)) {
				if (logger.isDebugEnabled()) {
					logger.debug("FSM[" + name + "] busy, event[" + event
							+ "] has been dropped.");
				}
				return;
			}

			final MutableFSMContext initCtx = contextFactory.createFSMContext(
					this, event);
			if (null == initCtx) {
				logger.warn("Can not create context for event " + event
						+ " , just ignore");
				return;
			}

			incHandledTransactionStart();

			ExecutorService service = this.selectService();

			if (null == service) {
				throw new RuntimeException(
						"Internal Error: can not select a valid executor service.");
			}

			// init ctx, and continue
			initCtx.setService(service);
			initCtx.setKey(key);

			if (null == contextHolder.put(key, initCtx)) {
				// add new
				ctxCount.incrementAndGet();
			}

			service.submit(new Runnable() {

				public void run() {
					initContext(initCtx, event);
				}
			});
		} else {
			ctx.getService().submit(new Runnable() {

				public void run() {
					dispatchEvent(ctx, event);
				}
			});
		}
	}

	// in timerExecService thread
	private void doScheduleTimeout(TimeoutEvent event, long delay) {
		timerEvents.put(event, System.currentTimeMillis() + delay);
	}

	// in timerExecService thread
	private void doCancelTimeout(TimeoutEvent event) {
		if (null == timerEvents.remove(event)) {
			if (logger.isWarnEnabled()) {
				logger.warn("Can not found matched timeout event[" + event
						+ "], maybe has been emited.");
			}
		}
	}

	private void scheduleCheckTimeoutEvents() {
		this.timerExecService.schedule(new Runnable() {

			public void run() {
				doCheckTimeoutEvents();
			}
		}, this.timerEventCheckInterval, TimeUnit.MILLISECONDS);
	}

	// in timerExecService thread
	private void doCheckTimeoutEvents() {
		long now = System.currentTimeMillis();

		try {
			Set<Map.Entry<TimeoutEvent, Long>> entrySet = timerEvents
					.entrySet();

			Iterator<Map.Entry<TimeoutEvent, Long>> iter = entrySet.iterator();
			while (iter.hasNext()) {
				Map.Entry<TimeoutEvent, Long> entry = iter.next();
				if (entry.getValue() <= now) {
					// condition matched, emit timeout event
					iter.remove();
					this.acceptEvent(entry.getKey());
				}
			}
		} catch (Exception e) {
			logger.error("doCheckTimeoutEvents:", e);
		} finally {
			scheduleCheckTimeoutEvents();
		}
	}

	public Runnable scheduleTimeout(final TimeoutEvent event, final long delay) {
		this.timerExecService.submit(new Runnable() {

			public void run() {
				doScheduleTimeout(event, delay);
			}
		});

		return new Runnable() {

			public void run() {
				timerExecService.submit(new Runnable() {

					public void run() {
						doCancelTimeout(event);
					}
				});
			}
		};
	}

	private int getPendingTaskCountOf(ExecutorService exec) {
		if (exec instanceof ThreadPoolExecutor) {
			BlockingQueue<Runnable> queue = ((ThreadPoolExecutor) exec)
					.getQueue();
			return queue.size();
		} else {
			throw new RuntimeException(
					"Internal Error-EventReactor : exec is !NOT! ScheduledThreadPoolExecutor class");
		}
	}

	public int[] getPendingTaskCount() {
		int[] ret = new int[this.execServices.length];
		int idx = 0;
		for (ExecutorService exec : this.execServices) {
			ret[idx++] = getPendingTaskCountOf(exec);
		}

		return ret;
	}

	/**
	 * @return the initeventAcceptor
	 */
	public String getIniteventAcceptorAsString() {
		return initeventAcceptor.toString();
	}

	public Predicate getIniteventAcceptor() {
		return initeventAcceptor;
	}

	/**
	 * @param initeventAcceptor
	 *            the initeventAcceptor to set
	 */
	public void setIniteventAcceptor(Predicate initeventAcceptor) {
		this.initeventAcceptor = initeventAcceptor;
	}

	public Closure getBusyHandler() {
		return busyHandler;
	}

	public void setBusyHandler(Closure busyHandler) {
		this.busyHandler = busyHandler;
	}

	public boolean isLogStateChanged() {
		return logStateChanged;
	}

	public void setLogStateChanged(boolean logStateChanged) {
		this.logStateChanged = logStateChanged;
	}

	public boolean isCheckBusy() {
		return checkBusy;
	}

	public void setCheckBusy(boolean checkBusy) {
		this.checkBusy = checkBusy;
	}

	public int getBusyThreshold() {
		return busyThreshold;
	}

	public void setBusyThreshold(int busyThreshold) {
		this.busyThreshold = busyThreshold;
	}

	/**
	 * @return the handledTransactionCount
	 */
	public int getHandledTransactionCount() {
		return handledTransactionCount.get();
	}

	/**
	 * @return the bypassTransactionCount
	 */
	public int getBypassTransactionCount() {
		return bypassTransactionCount.get();
	}

	/**
	 * @return the statisticer
	 */
	public FSMStatisticer getStatisticer() {
		return statisticer;
	}

	/**
	 * @param statisticer
	 *            the statisticer to set
	 */
	public void setStatisticer(FSMStatisticer statisticer) {
		this.statisticer = statisticer;
	}

	/**
	 * @return the fsmId
	 */
	public int getFsmId() {
		return fsmId;
	}

	/**
	 * @return the timerEventCheckInterval
	 */
	public long getTimerEventCheckInterval() {
		return timerEventCheckInterval;
	}

	/**
	 * @param timerEventCheckInterval
	 *            the timerEventCheckInterval to set
	 */
	public void setTimerEventCheckInterval(long timerEventCheckInterval) {
		this.timerEventCheckInterval = timerEventCheckInterval;
	}

	public int getPendingTimeoutEvent() {
		return this.timerEvents.size();
	}

	public int getPendingTimerTaskCount() {
		if (this.timerExecService instanceof ScheduledThreadPoolExecutor) {
			BlockingQueue<Runnable> queue = ((ScheduledThreadPoolExecutor) timerExecService)
					.getQueue();
			return queue.size();
		} else {
			throw new RuntimeException(
					"Internal Error-EventReactor : timerExecService is !NOT! ScheduledThreadPoolExecutor class");
		}
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString() {
		return this.name;
	}
}
